package com.zheng.flink.study.dataset.transformation;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Tuple;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by zhengbo on 2019/12/13.
 */
public class FlatMapTransformation {

    public static void main(String[] args) throws Exception {

        List<String> wordList = new ArrayList<>();

        wordList.add("hadoop,flink");
        wordList.add("hadoop,spark");
        wordList.add("storm,yarn");

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> dataSource = env.fromCollection(wordList);

        //通过flatMap和map计算单词出现次数
        dataSource.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String input, Collector<String> collector) throws Exception {

                for (String value : input.split(",")) {
                    collector.collect(value);
                }
            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {

            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value, 1);
            }
        }).groupBy(0).sum(1).print();


        //直接通过flatMap实现
        dataSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String input, Collector<Tuple2<String, Integer>> collector) throws Exception {
                for (String value : input.split(",")) {
                    collector.collect(new Tuple2<>(value, 1));
                }
            }
        }).groupBy(0).sum(1).print();


        dataSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String input, Collector<Tuple2<String, Integer>> collector) throws Exception {
                for (String value : input.split(",")) {
                    collector.collect(new Tuple2<>(value, 1));
                }
            }
        }).groupBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }).first(10).print();
    }
}
